package com.tbdp.lsmp.processor;

import com.tbdp.lsmp.common.Configuration;
import com.tbdp.lsmp.common.Constants;
import com.tbdp.lsmp.msg.StockMsg;
import com.tbdp.lsmp.util.JsonUtil;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;

/**
 * kafka writer
 *
 * @author xbkaishui
 * @version $Id: KafkaProcess.java, v 0.1 2017-11-05 下午11:08 xbkaishui Exp $$
 */
public class KafkaProcessor {

    private static Logger logger = LoggerFactory.getLogger(KafkaProcessor.class);

    private KafkaProducer producer;

    private String topic;

    public void init() {

        String kafkaBrokers = Configuration.getInstance().getProperty(Constants.KAFKA_BROKERS);
        topic = Configuration.getInstance().getProperty(Constants.TOPIC);

        logger.info("kafka brokers {} ", kafkaBrokers);
        logger.info("kafka topic {} ", topic);

        HashMap<String, Object> props = new HashMap<String, Object>();
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, "1000");
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokers);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        producer = new KafkaProducer(props);
    }

    public void process(Object msg) {
        StockMsg stockMsg = new StockMsg();
        stockMsg.setMsgType(3);
        stockMsg.setMsg(msg);
        String value = JsonUtil.toJsonString(stockMsg);
        logger.info("send msg to kafka {} ", value);
        ProducerRecord producerRecord = new ProducerRecord(topic, value);
        producer.send(producerRecord);
    }

    public KafkaProducer getProducer() {
        return producer;
    }

    public String getTopic() {
        return topic;
    }
}
